package cc.xpcas.raft.context;

import java.util.*;
import java.util.concurrent.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cc.xpcas.raft.message.*;
import cc.xpcas.raft.role.Role;

/**
 * @author xp
 */
public class MessageHandler {

    private static final Logger LOG = LoggerFactory.getLogger(MessageHandler.class);

    private final Context context;

    private Queue<MessageContext> queue = new LinkedBlockingQueue<>();

    public MessageHandler(Context context) {
        this.context = context;
    }

    public void enqueue(MessageContext messageContext) {
        enqueue(messageContext, false);
    }

    private void enqueue(MessageContext messageContext, boolean requeue) {
        if (!requeue) {
            LOG.info(String.format("%s: enqueue %s",
                    context.getConfiguration().getServerId(),
                    messageContext.getMessage().getClass().getSimpleName()
            ));
        } else {
            LOG.info(String.format("%s: requeue %s",
                    context.getConfiguration().getServerId(),
                    messageContext.getMessage().getClass().getSimpleName()
            ));
        }

        queue.add(messageContext);
    }

    private MessageContext dequeue() {
        return queue.poll(); // return null if empty
    }

    public boolean runOnce() throws InterruptedException {
        MessageContext messageContext = dequeue();
        if (messageContext == null) {
            //LOG.info(String.format("%s: no message", context.getConfiguration().getServerId()));
            return false;
        }

        if (!dispatch(messageContext)) {
            enqueue(messageContext, true);
        }

        // continuable
        return true;
    }

    private boolean dispatch(MessageContext messageContext) throws InterruptedException {
        LOG.info(String.format("%s: dispatch %s",
                context.getConfiguration().getServerId(),
                messageContext.getMessage().getClass().getSimpleName()
        ));

        Message message = messageContext.getMessage();
        Role role = context.getRole();
        if (message instanceof AppendEntriesRequest) {
            return role.handleAppendEntriesRequest(messageContext);
        } else if (message instanceof AppendEntriesResponse) {
            return role.handleAppendEntriesResponse(messageContext);
        } else if (message instanceof RequestVoteRequest) {
            return role.handleRequestVoteRequest(messageContext);
        } else if (message instanceof RequestVoteResponse) {
            return role.handleRequestVoteResponse(messageContext);
        } else {
            if (LOG.isWarnEnabled()) {
                LOG.warn(String.format("unhandled message: %s", messageContext.getClass().getName()));
            }
            return true;
        }
    }
}
